package org.hbhk.hms.consumer;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.hbhk.hms.mq.MqConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

/**
 * 
 * 消费处理，对应的topic未触发数据
 *
 */
@Service
public class HmsTopicDataTestConsumer implements InitializingBean {

	private Logger log = LoggerFactory.getLogger(getClass());

	public String subscribe = "hbhk2";

	@Value("${mq.reconsumeTimes:3}")
	public Integer reconsumeTimes = 3;

	private DefaultMQPushConsumer consumer;

	@Autowired
	private MqConsumer mqConsumer;

	public void init() throws Exception {

		log.error("HmsDelalyData consumer start...");

		consumer = mqConsumer.get("HmsDelayDataTest_consumer");

		consumer.subscribe(subscribe, "*");

		consumer.registerMessageListener(new MessageListenerConcurrently() {

			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
				MessageExt me = msgs.get(0);
				try {
					String topic = me.getTopic();
					String tags = me.getTags();
					String keys = me.getKeys();
					String msgBody = new String(me.getBody());
					log.error("topic: " + topic + ",tags: " + tags + ",key: " + keys + ",msgBody: " + msgBody);
				} catch (Exception e) {
					log.error(e.getMessage(), e);
					int reconsumeTimes = me.getReconsumeTimes();
					log.error("reconsumeTimes: " + reconsumeTimes);
					if (reconsumeTimes >= reconsumeTimes) {
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
					return ConsumeConcurrentlyStatus.RECONSUME_LATER;
				}
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		});

		consumer.start();
		log.error("HmsDelalyData consumer end...");

	}

	@Override
	public void afterPropertiesSet() throws Exception {
		init();
	}

}
